package master;

import java.io.BufferedInputStream;
import java.io.BufferedOutputStream;
import java.io.BufferedReader;
import java.io.ByteArrayOutputStream;
import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.EOFException;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.io.FileReader;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.io.StringReader;
import java.net.DatagramPacket;
import java.net.DatagramSocket;
import java.net.InetAddress;
import java.net.ServerSocket;
import java.net.Socket;
import java.net.SocketException;
import java.net.UnknownHostException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.Date;
import java.util.Hashtable;
import java.util.Iterator;
import java.util.List;
import java.util.ListIterator;
import java.util.PriorityQueue;

import java.util.logging.Level;
import java.util.logging.Logger;
import javax.xml.bind.JAXBContext;
import javax.xml.bind.JAXBException;
import javax.xml.bind.Marshaller;
import javax.xml.bind.Unmarshaller;
import javax.xml.parsers.DocumentBuilder;
import javax.xml.parsers.DocumentBuilderFactory;
import javax.xml.parsers.ParserConfigurationException;
import javax.xml.transform.stream.StreamSource;

import org.w3c.dom.Document;
import org.w3c.dom.Element;
import org.w3c.dom.Node;
import org.w3c.dom.NodeList;

import common.ClientObject;
import common.FileObject;
import common.JobObject;
import common.JobRequestObject;
import common.JobResultObject;
import common.JobsPackage;
import common.MasterObject;
import common.Util;
import common.WorkerObject;

public class MasterService {

	private InetAddress serverIP;
	private int serverTCPPort;
	private MasterObject master;
	private ArrayList<WorkerObject> workerList = new ArrayList<WorkerObject>();
	private ArrayList<String> resultList = new ArrayList<String>();
	private PriorityQueue<JobObject> jobQueue;
	private PriorityQueue<JobsPackage> jobsPackageQueue;
	private int reveiveJobResultsTCPPort = 5002;
	public Hashtable<String, ClientObject> clientHashtable = new Hashtable<String, ClientObject>();
	private int clientCounter = 0;
	//keep track of jobs status
	public Hashtable<String, JobsPackage> jobPackageInfo = new Hashtable<String, JobsPackage>();
	public Hashtable<Integer, WorkerObject> workerInfo = new Hashtable<Integer, WorkerObject>();
	static String DEAD = "DEAD";
	static String AVAILABLE = "AVAILABLE";
	static String RUNNING = "RUNNING";
	static final String WAITING = "WAITING";
	static String COMPLETED = "COMPLETED";
	String FAILED = "FAILED";
    static String FAIL = "FAIL";
	private int workerIDCounter = 0;

	public MasterService() {
	}

	public MasterService (MasterObject master, ArrayList<WorkerObject> workerList, PriorityQueue<JobsPackage> jobsPackageQueue, PriorityQueue<JobObject> jobQueue) {
		this.master = master;
		this.workerList = workerList;
		this.jobQueue = jobQueue;
		this.jobsPackageQueue = jobsPackageQueue;
	}

	public MasterService (InetAddress aServerIP, int aServerTCPPort, ArrayList<WorkerObject> aWorkerList, ArrayList<String> aResultList) {
		this.serverIP = aServerIP;
		this.serverTCPPort = aServerTCPPort;
		this.workerList = aWorkerList;
		this.resultList = aResultList;
	}

	public void register(MasterService masterService){

		System.out.println("SYSINFO - Register Service start.");
		try {
			System.out.println("SYSINFO - Register Service works on address: "+InetAddress.getLocalHost().getHostAddress()+":"+master.getMasterUDPPort());
		} catch (UnknownHostException e1) {
			// TODO Auto-generated catch block
			e1.printStackTrace();
		}

		DatagramSocket aSocket = null;

		try {
			aSocket = new DatagramSocket(master.getMasterUDPPort());
			byte buf[] = new byte[1024];
			Long serverLocalTime = null;

			while (true) {
				System.out.println("SYSINFO - Waiting for registration request.");
				DatagramPacket request = new DatagramPacket(buf, buf.length);
				aSocket.receive(request);
				serverLocalTime = getTime();
				System.out.println("SYSINFO - Received registration request.");
				String registerXML = new String(buf, 0, request.getLength());
				//				System.out.println("DEBUG - XML Request: "+regRequest);

				JAXBContext context = JAXBContext.newInstance(RegisterRequestObject.class);
				Unmarshaller u = context.createUnmarshaller();
				StringBuffer xmlStr = new StringBuffer(registerXML);
				RegisterRequestObject registerRequest = (RegisterRequestObject) u.unmarshal( new StreamSource( new StringReader( xmlStr.toString() ) ) );

				System.out.println("Registration request information:");
				if (registerRequest.getFrom().equals("Worker")&&registerRequest.getType().equals("Register")) {
					WorkerObject worker = registerRequest.getWorker();

					System.out.println("DEBUG - ip = " + worker.getIp());
					System.out.println("DEBUG - port = " + worker.getPort());
					System.out.println("DEBUG - price = " + worker.getPrice());

					System.out.println("SYSINFO - Check Worker list before register a new Worker address.");
					if (workerList.isEmpty()) {
						//						System.out.println("DEBUG - Worker list is empty.");
						workerList.add(worker);
						worker.setWorkerID(workerIDCounter++);
						worker.setRegTime(serverLocalTime);
						worker.setStatus(AVAILABLE);
						
						workerInfo.put(worker.getWorkerID(), worker);
						
						System.out.println("SYSINFO - Register the new Worker address.");
						System.out.println("DEBUG - time = " + worker.getWorkerID());
						System.out.println("DEBUG - time = " + worker.getRegTime());
						System.out.println("DEBUG - time = " + worker.getStatus());
						
						// start up a new worker handling thread when every new worker registered on master
						WorkerHandlingThread workerHandlingThread = new WorkerHandlingThread(worker, masterService);
						workerHandlingThread.start();

					} else {
						//	System.out.println("DEBUG - Worker list is nonempty. Start to scan the list......");
						ListIterator<WorkerObject> iter = workerList.listIterator();
						boolean sameIndicator = false;
						while (iter.hasNext()) {
							WorkerObject workerTemp = iter.next();
							if (workerTemp.getIp().equals(worker.getIp()) && workerTemp.getPort() == worker.getPort()) {
								System.out.println("SYSINFO - Found a duplicated Worker address: "
										+workerTemp.getRegTime()
										+ " / " + workerTemp.getIp() + ":"
										+ workerTemp.getPort());
								workerTemp.setRegTime(serverLocalTime);
								iter.set(workerTemp);
								workerInfo.get(workerTemp.getWorkerID()).setRegTime(serverLocalTime);
								System.out.println("SYSINFO - Renew the register time for the duplicated Worker address.");
								System.out.println("SYSINFO - Renewed Worker information: "
										+ workerTemp.getRegTime()
										+ " / " + workerTemp.getIp() + ":"
										+ workerTemp.getPort());
								sameIndicator = true;
							}
						}
						if (sameIndicator == false) {
							//							System.out.println("DEBUG - No duplicated Worker address was found.");
							WorkerObject newWorker = new WorkerObject();
							newWorker.setIp(worker.getIp());
							newWorker.setPort(worker.getPort());
							newWorker.setPrice(worker.getPrice());
							newWorker.setLoad(worker.getLoad());
							newWorker.setWorkerID(workerIDCounter++);
							newWorker.setRegTime(serverLocalTime);
							newWorker.setStatus(AVAILABLE);
							iter.add(newWorker);
							
							workerInfo.put(newWorker.getWorkerID(), newWorker);
							
							System.out.println("SYSINFO - Register the new Worker address.");
							System.out.println("DEBUG - time = " + newWorker.getWorkerID());
							System.out.println("DEBUG - time = " + newWorker.getRegTime());
							System.out.println("DEBUG - time = " + newWorker.getStatus());
							
							// start up a new worker handling thread when every new worker registered on master
							WorkerHandlingThread workerHandlingThread = new WorkerHandlingThread(newWorker, masterService);
							workerHandlingThread.start();
						}
					}
				} else {
					System.out.println("SYSINFO - Registration Request is not from Worker.");
				}
			}
		} catch (SocketException e){System.out.println("Socket:" + e.getMessage());
		} catch (IOException e){System.out.println("IO:" + e.getMessage());
		} catch (JAXBException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		} finally {
			if(aSocket != null) {
				aSocket.close();
				System.out.println("SYSINFO - Client TCP connection is close.");
			}
		}
	}

	public void incrementClientNumber() {
		clientCounter++;
	}

	public int getClientNumber() {
		return clientCounter;
	}

	public void workerTimeout() {
		Long currentTime = getTime();
		ListIterator<WorkerObject> iter = workerList.listIterator();
		WorkerObject worker = null;
		while (iter.hasNext()) {
			worker = iter.next();
			if ((currentTime - worker.getRegTime()) > 3000) {
				worker.setStatus(DEAD);
				// worker is DEAD
				iter.remove();
				System.out.println("SYSINFO - Worker at address "+worker.getIp()+":"+worker.getPort()+" is dead.");
				ArrayList<JobObject> jobList = worker.getJobsList();
				System.out.println("SYSINFO - Rescheduling worker jobs. Number of Jobs: " + jobList.size());
				RescheduleWorkerJobs rescheduler = new RescheduleWorkerJobs(this, workerList, jobList);
				rescheduler.start();
			}
		}
		System.out.println("SYSINFO - Worker timeout checking is finished.");
	}

	// requesting worker list from Cloud Server
	public void requestList(){
		//		SSLSocketFactory sslsocketfactory = (SSLSocketFactory) SSLSocketFactory.getDefault();
		//      SSLSocket s = null;
		Socket s = null;
		try {
			//			s = (SSLSocket) sslsocketfactory.createSocket(serverIP, serverTCPPort);
			s = new Socket(serverIP, serverTCPPort);

			// create a list request and send it to Cloud Server
			OutputStream out = s.getOutputStream();
			DataOutputStream dataOut = new DataOutputStream(out);
			String listRequest = createListRequestXML();
			dataOut.writeUTF(listRequest);
			System.out.println("SYSINFO - Send a list request.");

			// receive a list reply from Cloud Server
			InputStream in = s.getInputStream();
			DataInputStream dataIn = new DataInputStream(in);
			String listReply = dataIn.readUTF();
			System.out.println("SYSINFO - Received a list reply.");
			//			System.out.println("DEBUG - Reply XML - "+listReply);

			// process the list reply and restore the available worker list
			Document inDoc = Util.stringToDoc(listReply);
			Element inRoot = inDoc.getDocumentElement();

			if (inRoot.getAttribute("type").equals("ListReply")) {
				//				System.out.println("DEBUG - The reply is ListReply.");
				NodeList nodeList = inRoot.getElementsByTagName("Worker");

				System.out.println("SYSINFO - List for available Workers:");
				for (int i = 0; i < nodeList.getLength(); i++) {
					Node node = nodeList.item(i);
					if (node.getNodeType() == Node.ELEMENT_NODE) {
						Element element = (Element) node;
						String ip = Util.getTagValue("IP", element);
						int port = Integer.parseInt(Util.getTagValue("Port", element));
						int price = Integer.parseInt(Util.getTagValue("Price", element));
						int load = Integer.parseInt(Util.getTagValue("Load", element));
						long localServerTime = getTime();
						System.out.println("SYSINFO - Worker("+i+") IP: " + ip);
						System.out.println("SYSINFO - Worker("+i+") Port: " + port);
						System.out.println("SYSINFO - Worker("+i+") Price: " + price);
						System.out.println("SYSINFO - Worker("+i+") Load: " + load);
						//TODO: connect to the worker
						WorkerObject worker = new WorkerObject();
						worker.setIp(ip);
						worker.setPort(port);
						worker.setPrice(price);
						worker.setLoad(load);
						worker.setRegTime(localServerTime);
						workerList.add(worker);
						// Connection c = new Connection(workerHost,workerPort);
					}
				}
			} else {
				System.out.println("SYSINFO - The reply is not ListReply. No action will be performed.");
			}
		} catch (UnknownHostException e) {System.out.println("Sock:"+e.getMessage());
		} catch (EOFException e) {System.out.println("EOF:"+e.getMessage());
		} catch (IOException e) {System.out.println("IO:"+e.getMessage());
		} catch (Exception e) {e.printStackTrace();
		} finally {
			if(s!=null) {
				try {
					s.close();
					System.out.println("SYSINFO - Client to Server TCP connection is close.");
				} catch(IOException e){/*close failed*/}
			}
		}
	}

	// create XML document for requesting List
	public String createListRequestXML() throws Exception {
		DocumentBuilderFactory dbfactory = DocumentBuilderFactory.newInstance();
		DocumentBuilder db = null;
		try {
			db = dbfactory.newDocumentBuilder();
		} catch (ParserConfigurationException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		Document outDoc = db.newDocument();
		outDoc.setXmlVersion("1.0");
		Element outRoot = outDoc.createElement("Client");
		outRoot.setAttribute("type", "List");
		outRoot.setAttribute("version", "1");

		Element timestampElement = outDoc.createElement("Timestamp");
		Long time = getTime();
		timestampElement.appendChild(outDoc.createTextNode(time.toString()));
		outRoot.appendChild(timestampElement);

		outDoc.appendChild(outRoot);

		String xmlStr = Util.docToString(outDoc);
		// System.out.println("DEBUG - XML Request: "+xmlStr);

		return xmlStr;
	}
	/*
	// receiving jobs from clients
	public void ReceiveJobPackages(PriorityQueue<JobObject> jobQueue, MasterService masterService) {

		System.out.println("SYSINFO - Worker ReceiveJob service start.");

		try{
			//			SSLServerSocketFactory sslserversocketfactory = (SSLServerSocketFactory) SSLServerSocketFactory.getDefault();
			//			SSLServerSocket listenSocket = (SSLServerSocket) sslserversocketfactory.createServerSocket(workerTCPPort);
			ServerSocket listenSocket = new ServerSocket(master.getMasterTCPPort());

			while (true) {
				Socket clientSocket = listenSocket.accept();
				//				SSLSocket clientSocket = (SSLSocket) listenSocket.accept();
				System.out.println("SYSINFO - Accept Client TCP connection for transfering a job.");

				ReceiveJobPackageThread c = new ReceiveJobPackageThread(clientSocket, jobQueue, masterService);
				c.start();
			}
		}catch(IOException e){System.out.println("Listen:"+e.getMessage());}
	}*/

	private void scheduleJobsByCost(JobsPackage jp){
		ArrayList<JobObject> jobList = jp.getJobs();
		System.out.println("SYSINFO - Assigning jobs from client to workers (Cost Minimization).");
		Collections.sort(workerList, new SortWorkers("PRICE", "ACS"));
		//System.out.println("DEBUG - Worker Seq 1: "+workerList.get(0).getWorkerID());
		//System.out.println("DEBUG - Worker Seq 1: "+workerList.get(0).getPrice());
		//System.out.println("DEBUG - Worker Seq 2: "+workerList.get(1).getWorkerID());
		//System.out.println("DEBUG - Worker Seq 2: "+workerList.get(1).getPrice());
		if (!jobList.isEmpty()&&!workerList.isEmpty()) {
			JobObject job = null;
			WorkerObject worker = workerList.get(0); // set for same price worker
			long deadline = jp.getDeadline(); 
			for (int count = 0, n = 0; count < jobList.size(); count++, n++) {
				job = jobList.get(count);

				if(n>=workerList.size()) n = 0;
				if(!(workerList.get(n).getPrice() == worker.getPrice())) {
					n = 0;
					System.out.println("DEBUG - Worker ID: "+workerList.get(n).getWorkerID());
				}
				int i=0;
				//while (i < workerList.size()-1 && workerList.get(i).getLoad() + job.getTime() > deadline)
				//	i++;
				while (i < workerList.size() && checkTime(workerList.get(i), job, deadline))
					i++;
				if (i==workerList.size()) {
					System.out.println("SYSINFO - Unable to schedule all jobs within given time");
					i=0;
				}

				workerList.get(n).setJobsList(addJobToList(job, workerList.get(n).getJobsList(), jp.getDeadline()));
				
				System.out.println("DEBUG - No 0: "+workerList.get(n).getWorkerID());
				System.out.println("DEBUG - No 1: "+workerList.get(n).getJobsList().size());
                //record the assignment of job to which worker
                job.setAllocateWorkerID(workerList.get(n).getWorkerID());
                // update Worker Info
                workerInfo.put(workerList.get(n).getWorkerID(), workerList.get(n));
                // update the worker load
                workerList.get(n).setLoad(workerList.get(n).getLoad() + job.getTime());

				System.out.println("DEBUG - Job ID: " + job.getJobID());
				System.out.println("DEBUG - Job FileName: " + job.getExecuteFileName());
				System.out.println("DEBUG - Job Parameter: " + job.getParameters());
				System.out.println("DEBUG - Job Priority: " + job.getPriority());
			}
		} else {
			System.out.println("SYSINFO - Job/Worker list is empty.");
		}
		/*System.out.println("DEBUG - No 7: "+workerList.get(0).getWorkerID());
		System.out.println("DEBUG - No 8: "+workerList.get(0).getJobsList().size());
		System.out.println("DEBUG - No 9: "+workerInfo.get(0).getWorkerID());
		System.out.println("DEBUG - No 10: "+workerInfo.get(0).getJobsList().size());
		
		System.out.println("DEBUG - No 7: "+workerList.get(1).getWorkerID());
		System.out.println("DEBUG - No 8: "+workerList.get(1).getJobsList().size());
		System.out.println("DEBUG - No 9: "+workerInfo.get(1).getWorkerID());
		System.out.println("DEBUG - No 10: "+workerInfo.get(1).getJobsList().size());*/
	}

	public boolean checkTime(WorkerObject workerObject, JobObject job,
			long deadline) {
		ArrayList<JobObject> jobList = workerObject.getJobsList();
		int i=jobList.size()-1;
		if (i==-1){
			return false;
		}
		while (i>=0 && jobList.get(i).getExtraTime() >= job.getTime() 
				&& jobList.get(i).getPriority() < job.getPriority()) {
			i--;
		}
		int totalTime = 0;
		while (i>0) {
			totalTime += jobList.get(i).getTime();
			if (totalTime > deadline) return true;
			i--;
		}

		return false;
	}

	private void scheduleJobsByTime(JobsPackage jp){
		ArrayList<JobObject> jobList = jp.getJobs();
		System.out.println("SYSINFO - Assigning jobs from client to workers (Time Minimization).");
		if (!jobList.isEmpty()&&!workerList.isEmpty()) {
			JobObject job =null;
			int budget = jp.getBudget(); 
			int unscheduledJobs = jobList.size();
			for (int count = 0; count < jobList.size(); count++) {
				Collections.sort(workerList, new SortWorkers("TIME", "ACS"));
				job = jobList.get(count);
				int estimatedExecTime = job.getTime();
				int i=0;
				//while (workerList.get(i).getPrice()*estimatedExecTime > budget/unscheduledJobs && i < workerList.size()-1)
				//	i++;
				while (i < workerList.size() && checkCost(workerList.get(i), job, budget/unscheduledJobs))
					i++;
				if (i==workerList.size()) {
					System.out.println("SYSINFO - Unable to schedule all jobs within given budget");
					i=0;
				}
				workerList.get(i).setJobsList(addJobToList(job, workerList.get(i).getJobsList(), jp.getDeadline()));
                //record the assignment of the job to which worker
                job.setAllocateWorkerID(workerList.get(i).getWorkerID());
				workerList.get(i).setLoad(workerList.get(i).getLoad() + job.getTime());
				unscheduledJobs--;
				budget -= workerList.get(i).getPrice()*estimatedExecTime;

				System.out.println("DEBUG - Job ID: "
						+ job.getJobID());
				System.out.println("DEBUG - Job FileName: "
						+ job.getExecuteFileName());
				System.out.println("DEBUG - Job Parameter: "
						+ job.getParameters());
				System.out.println("DEBUG - Job Priority: "
						+ job.getPriority());


			}
		} else {
			System.out.println("SYSINFO - Job/Worker list is empty.");
		}
	}

	public boolean checkCost(WorkerObject workerObject,
			JobObject job, int budgetPerJob) {
		if (budgetPerJob > workerObject.getPrice()*job.getTime()) return false;
		else return true;
	}

	public ArrayList<JobObject> addJobToList(JobObject job, ArrayList<JobObject> jobList, long deadline) {

		int i=jobList.size()-1;
		if (i<0){
			job.setExtraTime((int) (deadline - job.getTime()));
/*			else job.setExtraTime((int) (deadline - job.getTime() - jobList.get(0).getTime()));*/
			jobList.add(job);
			return jobList;
		}
		else {
			while (i>=0 && jobList.get(i).getExtraTime() >= job.getTime() 
					&& jobList.get(i).getPriority() < job.getPriority()) {
				i--;
			}
			i++;
			jobList.add(i, job);
			int previousTimes = 0;
			for (int j=0; j<i; j++)
				previousTimes+=jobList.get(i).getTime();
			job.setExtraTime((int) (deadline - job.getTime() - previousTimes));
			i++;
			while (i < jobList.size()) {
				jobList.get(i).setExtraTime(jobList.get(i).getExtraTime() - job.getTime());
				jobList.set(i, jobList.get(i));
				i++;
			}
			//worker.setJobsList(jobList);
			return jobList;
		}
		
/*		int i=jobList.size()-1;
		if (i<1){
			if (i<0) job.setExtraTime((int) (deadline - job.getTime()));
			else job.setExtraTime((int) (deadline - job.getTime() - jobList.get(0).getTime()));
			jobList.add(job);
			return jobList;
		}
		else {
			while (i>=1 && jobList.get(i).getExtraTime() >= job.getTime() 
					&& jobList.get(i).getPriority() < job.getPriority()) {
				i--;
			}
			i++;
			jobList.add(i, job);
			int previousTimes = 0;
			for (int j=0; j<i; j++)
				previousTimes+=jobList.get(i).getTime();
			job.setExtraTime((int) (deadline - job.getTime() - previousTimes));
			i++;
			while (i < jobList.size()) {
				JobObject thisJob = jobList.get(i);
				jobList.remove(i);
				thisJob.setExtraTime(thisJob.getExtraTime() - job.getTime());
				jobList.add(i, thisJob);
				i++;
			}
			//worker.setJobsList(jobList);
			return jobList;
		}*/

	}

	// assigning jobs from client to workers
	public void assignJobs(){

		while (true) {
			try {
				Thread.sleep(1000);
			} catch (InterruptedException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
			if (!jobsPackageQueue.isEmpty()&&!workerList.isEmpty()){
				JobsPackage jp = jobsPackageQueue.poll();
				ArrayList<JobObject> jobList = jp.getJobs();
				System.out.println("SYSINFO - Assigning jobs from master to workers.");
				Collections.sort(workerList, new SortByLoad("ACS"));
				if (!jobList.isEmpty()&&!workerList.isEmpty()) {
					if (jp.getOptimizeType().matches("TIME")) scheduleJobsByTime(jp);
					else scheduleJobsByCost(jp);
					for (int j=0; j<workerList.size(); j++) {
						ArrayList<JobObject> tempList = workerList.get(j).getJobsList();
						System.out.print("DEBUG - Worker Queue " + j + " : ");
						for (int i=0; i<tempList.size(); i++)
							System.out.print(tempList.get(i).getClientID() + "." + tempList.get(i).getJobID() + ";");
						System.out.println();
					}

					System.out.println("DEBUG - Nooooooooooooooo 1: "+workerInfo.get(workerList.get(0).getWorkerID()).getJobsList().isEmpty());
					System.out.println("DEBUG - Nooooooooooooooo 2: "+workerInfo.get(workerList.get(0).getWorkerID()).getJobsList().size());
					//add to JPInfo
					//for (int count = 0; count < jp.getJobs().size(); count++) {
					//	jp.getJobs().get(count).setStatus(WAITING);
					//}
					//jobPackageInfo.put(jp.getClientID(), jp);
					//System.out.println("DEBUG - jobPackageInfo :"+jobPackageInfo.get(jp.getClientID()).getJobs().get(0).getStatus());

					
				} else {
					System.out.println("DEBUG - Job list is empty.");
				}
			}
		}
	}

	public void queueJobsPackage(JobsPackage jp) {
		jobsPackageQueue.add(jp);
		System.out.println("DEBUG - jobsPackageQueue size: "+jobsPackageQueue.size());
		return;
	}

	public Long getTime() {
		Date date = new Date();
		return date.getTime();
	}

	public void workerQueueHandling(WorkerObject worker) {
		while (true) {
//			System.out.println("DEBUG - Yes 1: "+worker.getJobsList().isEmpty());
			try {
				Thread.sleep(2000);
//				System.out.println("DEBUG - Yes 2: "+worker.getJobsList().isEmpty());
				/*				while (worker.getJobsList().isEmpty()||worker.getStatus().equals(RUNNING)) {
					synchronized(this) {try {
						System.out.println("DEBUG - Yes 3: "+worker.getJobsList().isEmpty());
						wait();
					} catch (InterruptedException e) {}}
				}*/
				System.out.println("DEBUG - Worker "+worker.getWorkerID()+" size: "+worker.getJobsList().size());
				if (!worker.getJobsList().isEmpty()&&workerInfo.get(worker.getWorkerID()).getStatus().equals(AVAILABLE)){
					JobObject job = worker.getJobFromJobsList(0);
					System.out.println("SYSINFO - Assigning a job from master to worker.");
					if(assignJob(job, worker)){
						worker.getJobFromJobsList(0).setStatus(RUNNING);
						// NTD: set back to AVAILABLE when the Job Result is back from Worker
						worker.setStatus(RUNNING);
						workerInfo.get(worker.getWorkerID()).setStatus(RUNNING);
					} else {
						// handle assignJob exception
						worker.setStatus(DEAD);
						workerInfo.get(worker.getWorkerID()).setStatus(DEAD);
					}
				} else if(!worker.getJobsList().isEmpty()&&workerInfo.get(worker.getWorkerID()).getStatus().equals(RUNNING)){
					System.out.println("DEBUG - Job list is NOT empty. Worker is running.");
				} else if(worker.getJobsList().isEmpty()&&workerInfo.get(worker.getWorkerID()).getStatus().equals(RUNNING)) {
					System.out.println("DEBUG - Job list is empty. Worker is running.");
				} else if(worker.getJobsList().isEmpty()) {
					System.out.println("DEBUG - Job list is empty.");
				} else if(worker.getStatus().equals(DEAD)) {
					System.out.println("DEBUG - Worker is dead.");break;
				}
				/*				//Toggle status.
				jobsListEmpty = true;
				//Notify producer that status has changed.
				notifyAll();*/

			} catch (Exception e) {
				e.printStackTrace();
			}
		}
	}

	public boolean assignJob(JobObject job, WorkerObject worker) {
		String workerIPStr = worker.getIp();
		InetAddress workerIP = null;
		try {
			workerIP = InetAddress.getByName(workerIPStr);
		} catch (UnknownHostException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
			return false;
		}
		int workerTCPPort = worker.getPort();

		DataOutputStream dataOut;
		// SSLSocketFactory sslsocketfactory = (SSLSocketFactory)
		// SSLSocketFactory.getDefault();
		// SSLSocket socket = null;
		Socket socket = null;
		try {
			// socket = (SSLSocket) sslsocketfactory.createSocket(workerIP,
			// workerTCPPort);
			socket = new Socket(workerIP, workerTCPPort);

			OutputStream out = socket.getOutputStream();
			dataOut = new DataOutputStream(out);

			//Start Cost Calculation
			/*ArrayList<JobObject> jobList = jobPackageInfo.get(job.getClientID()).getJobs();
			for (int i=0; i<jobList.size(); i++) {
				if (jobList.get(i).getJobID() == job.getJobID()) {
					jobList.get(i).setStartTime(getTime());
					jobList.get(i).setWorkerCost(worker.getPrice());
				}
			}
			*/
			/********** Update JobPackage Info **********/
			ArrayList<JobObject> jobs = jobPackageInfo.get(job.getClientID()).getJobs();
			Iterator<JobObject> jobIter = jobs.iterator();
			
			while (jobIter.hasNext()) {
				JobObject job1 = jobIter.next();
				if (job1.getJobID()==job.getJobID()) {
					//jobPackageInfo.get(jobResult.getJob().getClientID()).getJobs().get(i).setStatus(COMPLETED);
					//jobPackageInfo.get(jobResult.getJob().getClientID()).getJobs().get(i).setEndTime(getTime());
					job1.setWorkerCost(worker.getPrice());
					job1.setJobStartTime(getTime());
				}
				
			}
			/********** JobPackage Information Transfer **********/
			String jobRequestXML = createJobRequestXML(job);
			System.out.println("DEBUG - XML Request:");
			System.out.println(jobRequestXML);
			dataOut.writeUTF(jobRequestXML);

			/********** File Transfer **********/
			System.out.println("DEBUG - fileName = " + job.getExecuteFileName());
			System.out.println("DEBUG - parameters = " + job.getParameters());
			System.out.println("DEBUG - time = " + job.getTime());
			System.out.println("DEBUG - jobID = " + job.getJobID());
			System.out.println("DEBUG - priority = " + job.getPriority());

			String filePath = job.getExecuteFilePath()+"/"+job.getClientID()+"/"+job.getExecuteFileName()+".class";
			File file = new File(filePath);
			System.out.println("DEBUG - File length:" + (int) file.length());
			dataOut.writeLong(file.length());

			// send the file content
			DataInputStream fileIn = new DataInputStream(new BufferedInputStream(new FileInputStream(filePath)));
			int bufferSize = 8192;
			byte[] buf = new byte[bufferSize];

			int length = 0;
			while((length = fileIn.read(buf))!=-1) {
				dataOut.write(buf, 0, length);
			}
			dataOut.flush();
			System.out.println("SYSINFO - Send a Job File.");

		} catch (IOException e) {
			e.printStackTrace();
			return false;
		} catch (Exception e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
			return false;
		} finally {
			if(socket!=null) {
				try {
					socket.close();
					System.out.println("SYSINFO - Client to Worker TCP connection is close.");
				} catch(IOException e){e.printStackTrace();return false;}
			}
		}
		return true;
	}

	// create XML document for requesting Job
	public String createJobRequestXML(JobObject job) throws Exception {

		JAXBContext context = JAXBContext.newInstance(JobRequestObject.class);
		Marshaller m = context.createMarshaller();
		m.setProperty(Marshaller.JAXB_FORMATTED_OUTPUT, true);

		JobRequestObject jobRequest = new JobRequestObject();
		jobRequest.setFrom("Master");
		jobRequest.setType("SendJob");
		jobRequest.setTimestamp(getTime());
		jobRequest.setJobObject(job);

		ByteArrayOutputStream baos = new ByteArrayOutputStream();
		m.marshal(jobRequest, baos);
		String jobRequestXML = baos.toString();

		return jobRequestXML;
	}

	public void ReceiveJobResults(MasterService masterService) {
		System.out.println("SYSINFO - Master receive job results service start.");

		try {
			// SSLServerSocketFactory sslserversocketfactory = (SSLServerSocketFactory) SSLServerSocketFactory.getDefault();
			// SSLServerSocket listenSocket = (SSLServerSocket)
			// sslserversocketfactory.createServerSocket(workerTCPPort);
			ServerSocket listenSocket = new ServerSocket(reveiveJobResultsTCPPort);

			while (true) {
				Socket workerSocket = listenSocket.accept();
				// SSLSocket clientSocket = (SSLSocket) listenSocket.accept();
				System.out.println("SYSINFO - Accept Worker TCP connection for transfering a job result.");

				ReceiveAJobResultThread receiveAJobResultThread = new ReceiveAJobResultThread(workerSocket, masterService);
				receiveAJobResultThread.start();
			}
		} catch (IOException e) {
			System.out.println("Listen:" + e.getMessage());
		}
	}

	public void receiveAJobResult(Socket workerSocket) {

		System.out.println("SYSINFO - Waiting job result from worker.");
		System.out.println("SYSINFO - WorkerInfo Table: "+workerInfo.size());

		boolean jpCompletedIndcator = false;
		try {
			InputStream in = workerSocket.getInputStream();
			DataInputStream dataIn = new DataInputStream(in);

			String jobResultXML = dataIn.readUTF();
			System.out.println("SYSINFO - Received a job result.");
			//System.out.println("DEBUG - result XML: " + jobResultXML);

			/********** JobResult Information Transfer **********/
			JAXBContext context = JAXBContext.newInstance(JobResultObject.class);
			Unmarshaller u = context.createUnmarshaller();
			StringBuffer xmlStr = new StringBuffer(jobResultXML);
			JobResultObject jobResult = (JobResultObject) u.unmarshal(new StreamSource(new StringReader(xmlStr.toString())));

			ArrayList<FileObject> fileList = new ArrayList<FileObject>();
			if (jobResult.getFrom().equals("Worker") && jobResult.getType().equals("SendJobResult")) {

				System.out.println("DEBUG - jobID = " + jobResult.getJob().getJobID());
				System.out.println("DEBUG - priority = " + jobResult.getJob().getStatus());
				System.out.println("DEBUG - clientID = " + jobResult.getJob().getClientID());
				System.out.println("DEBUG - status = " + jobResult.getJob().getStatus());
				System.out.println("DEBUG - NumberOfOutputFiles = " + jobResult.getNumberOfOutputFiles());

				fileList.addAll(jobResult.getFiles());

				String filePath = "masterJobs"+"/"+jobResult.getJob().getClientID(); // default jobs folder (but can be assign to another one)
				System.out.println(filePath);
				File file = new File(filePath);
				if (file.exists()==false) {
					file.mkdirs();
				}

				/********** File Transfer **********/
				String savePath = filePath;
				int bufferSize = 8192;
				byte[] buf = new byte[bufferSize];

				Iterator fileIter = fileList.iterator();
				FileObject fileObject = new FileObject();

				String tempPath;
				int length = 0;
				while ((length = dataIn.read(buf)) != -1) {
					System.out.println("DEBUG - Read data from connection into buffer.");

					int bufRead = 0;

					while (bufRead != bufferSize && fileIter.hasNext()) {
						System.out.println("DEBUG - Read data from buffer.");
						if (fileObject.isCompleted() && fileIter.hasNext()){
							System.out.println("DEBUG - A new File.");
							fileObject = (FileObject) fileIter.next();
						}

						if (fileObject.getFileSize() == 0) {
							System.out.println("DEBUG - File size is 0.");
							System.out.println("SYSINFO - File saved at " + fileObject.getFileName());
							tempPath = savePath + "/"+ fileObject.getFileName();
							DataOutputStream fileOut = new DataOutputStream(new BufferedOutputStream(new FileOutputStream(tempPath)));
							fileOut.write(buf, bufRead, (int) fileObject.getFileSize());
							fileOut.close();
							fileObject.setCompleted(true);
						}
						long befLeftLong = fileObject.getFileSize() - fileObject.getFileBufSave();
						int befLeft = (int) befLeftLong; // not safe casting

						if (bufferSize - bufRead >= befLeft && befLeft != 0) {
							System.out.println("DEBUG - Completed in one write operation. " + fileObject.getFileName());
							tempPath = savePath + "/" + fileObject.getFileName();
							DataOutputStream fileOut = new DataOutputStream(new BufferedOutputStream(new FileOutputStream(tempPath)));
							fileOut.write(buf, bufRead, (int) fileObject.getFileSize());
							fileOut.close();
							System.out.println("SYSINFO - File saved at " + fileObject.getFileName());
							fileObject.setFileBufSave(fileObject.getFileSize());
							bufRead += fileObject.getFileSize();
							fileObject.setCompleted(true);

						} else if (bufferSize - bufRead < befLeft) { // NTD: handle
							tempPath = savePath + "/" + fileObject.getFileName();
							DataOutputStream fileOut = new DataOutputStream(new BufferedOutputStream(new FileOutputStream(tempPath)));
							fileOut.write(buf, bufRead, bufferSize);
							System.out.println("DEBUG - File transfer haven't completed");
							fileObject.setFileBufSave(bufferSize - bufRead);
							fileOut.close(); // file has been written out, Master may scheduler an uncompleted file
							bufRead = bufferSize;
							fileObject.setCompleted(false);

						}
					}

				}
				
				/********** Update JobPackage Info **********/
				ArrayList<JobObject> jobs = jobPackageInfo.get(jobResult.getJob().getClientID()).getJobs();
				Iterator<JobObject> jobIter = jobs.iterator();
				//int i = 0;
				//jpCompletedIndcator = true;
				while (jobIter.hasNext()) {
					JobObject job = jobIter.next();
					if (job.getJobID()==jobResult.getJob().getJobID()) {
						//jobPackageInfo.get(jobResult.getJob().getClientID()).getJobs().get(i).setStatus(COMPLETED);
						//jobPackageInfo.get(jobResult.getJob().getClientID()).getJobs().get(i).setEndTime(getTime());
						job.setStatus(COMPLETED);
						job.setJobEndTime(getTime());
						System.out.println("DEBUG - Job Running Time : " + (job.getJobEndTime() - job.getJobStartTime()));
						break;
					}
/*					if (!job.getStatus().equals(COMPLETED)) {
						jpCompletedIndcator = false;
					}*/
					//i++;
				}
				/*
				// if all the jobs of a jobPackage are completed, notify the client.
				if (jpCompletedIndcator) {
					jobPackageInfo.get(jobResult.getJob().getClientID()).setJpCompletedIndcator(true);
					notifyClient(jobResult.getJob());
					System.out.println("SYSINFO - All Jobs in package Complete.");
				}
				*/
				if (jobPackageInfo.get(jobResult.getJob().getClientID()).getNumberOfJobsCompleted() +  
						jobPackageInfo.get(jobResult.getJob().getClientID()).getNumberOfJobsFailed()==
					jobPackageInfo.get(jobResult.getJob().getClientID()).getNumberOfJobs()) {
					jobPackageInfo.get(jobResult.getJob().getClientID()).setJpCompletedIndcator(true);
					jobPackageInfo.get(jobResult.getJob().getClientID()).setEndTime(getTime());
					notifyClient(jobResult.getJob());
					System.out.println("SYSINFO - All Jobs in package Complete.");
				}
				/********** Update Worker Info **********/
				System.out.println("SYSINFO - WorkerInfo Table: "+workerInfo.size());
				String ip = jobResult.getWorker().getIp();
				int port = jobResult.getWorker().getPort();
				System.out.println("DEBUG - Update Worker Info from 1"+ip+":"+port);
				System.out.println("DEBUG - Update Worker Info from 2"+workerList.get(0).getIp()+":"+workerList.get(0).getPort());
				System.out.println("DEBUG - workerList size: "+workerList.size());
				ListIterator<WorkerObject> iter = workerList.listIterator();
				while (iter.hasNext()) {
					WorkerObject workerTemp = iter.next();
					if (workerTemp.getIp().equals(ip) && workerTemp.getPort() == port) {
						System.out.println("SYSINFO - Found the Worker.");
						System.out.println("DEBUG - Worker Job List size B: "+workerTemp.getJobsList().size());
						workerTemp.getJobsList().remove(0);
						System.out.println("DEBUG - Worker Job List size A: "+workerTemp.getJobsList().size());
						workerTemp.setStatus(AVAILABLE);
						System.out.println("DEBUG - Worker("+workerTemp.getWorkerID()+") Status: "+workerInfo.get(workerTemp.getWorkerID()).getStatus());
					}
				}
				
				// print the job result back to screen of master
				printOutResult(jobResult.getJob());
				
			} else {
				System.out.println("SYSINFO - The reply is not a Job Result. No action will be performed.");
			}

			workerSocket.close();
			System.out.println("SYSINFO - Client TCP connection is close.");
		} catch (EOFException e) {
			System.out.println("EOF:" + e.getMessage());
		} catch (IOException e) {
			System.out.println("IO:" + e.getMessage());
		} catch (Exception e) {
			e.printStackTrace();
		} finally {
			try {
				if (workerSocket != null)
					workerSocket.close();
			} catch (IOException e) {/* close failed */
			}
		}
	}

	// print result to screen
	public void printOutResult(JobObject job) {

        System.out.println("JobID =" + job.getJobID());
        String jobID = Integer.toString(job.getJobID());
        String outputFileName = job.getExecuteFileName() + ".out";
        String errorFileName = job.getExecuteFileName() + ".err";
        String outputFilePath = "masterJobs/" + job.getClientID()+ "/" + outputFileName;
        String errorFilePath = "masterJobs/" + job.getClientID() + "/" + errorFileName;
        File outputFile = new File(outputFilePath);
        File errorFile = new File(errorFilePath);

        FileReader fr = null;
        //start to read .out file
        System.out.println("\nSYSINFO - Job Result Info from Client("+job.getClientID()+") - Job("+job.getJobID()+"):");
        if (outputFile.exists()) {
            try {
                fr = new FileReader(outputFilePath);
                BufferedReader br = new BufferedReader(fr);
                String testPrint;
                System.out.println("SYSINFO - Job Output Info :\n");
                while (br.ready()) {
                    testPrint = br.readLine(); //read one line
                    System.out.println(testPrint); //print the line in Console
                }
                System.out.println();
            } catch (IOException ex) {
                ex.printStackTrace();
                System.out.println("SYSINFO - Error to read the File in :" + outputFilePath);
            } finally {
                try {
                    fr.close();
                } catch (IOException ex) {
                    ex.printStackTrace();
                }
            }
        } else {
            System.out.println("SYSINFO - There are no such .out file");
        }
        
        if (errorFile.exists()) {
            try {
                fr = new FileReader(errorFilePath);
                BufferedReader br = new BufferedReader(fr);
                String testPrint;
                System.out.println("SYSINFO - Job Error Info :\n");
                while (br.ready()) {
                    testPrint = br.readLine(); //read one line
                    System.out.println(testPrint); //print the line in Console
                }
                System.out.println();
            } catch (IOException ex) {
                ex.printStackTrace();
                System.out.println("SYSINFO - Error to read the File in :" + errorFilePath);
            } finally {
                try {
                    fr.close();
                } catch (IOException ex) {
                    ex.printStackTrace();
                }
            }
        } else {
            System.out.println("SYSINFO - There are no such .err file");
        }
	}

	public void notifyClient(JobObject job) {
		Socket socket = null;
		try {
			ClientObject client = clientHashtable.get(job.getClientID());
			//			s = (SSLSocket) sslsocketfactory.createSocket(serverIP, serverTCPPort);
			socket = new Socket(client.getClientIP(), client.getClientRevPort());

			DataOutputStream dataOut = new DataOutputStream(socket.getOutputStream());

			JAXBContext context = JAXBContext.newInstance(JobsPackage.class);
			Marshaller m = context.createMarshaller();
			m.setProperty(Marshaller.JAXB_FORMATTED_OUTPUT, true);

			JobsPackage jp = jobPackageInfo.get(client.getClientID());
			jp.setFrom("Master");
			jp.setType("Notify");
			jp.setTimestamp(getTime());

			ByteArrayOutputStream baos = new ByteArrayOutputStream();
			m.marshal(jp, baos);
			String jobsXML = baos.toString();

			//			String jobsXML = createJobRequestXML(jp);
			System.out.println("DEBUG - XML Request:");
			//System.out.println(jobsXML);

			dataOut.writeUTF(jobsXML);
			System.out.println("SYSINFO - Send a Job Package Information.");
		} catch (UnknownHostException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		} catch (IOException e) {
			// NTD: try to connect 3 times or more
			System.out.println("SYSINFO - Connection refused. Client may be NOT available.");
			//e.printStackTrace();
		} catch (JAXBException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}  finally {
			try {
				if (socket != null)
					socket.close();
			} catch (IOException e) {/* close failed */
			}
		}
	}
	
	public void sendBackResult(String clientID, DataOutputStream dataOut) {

		try {
			// socket = (SSLSocket) sslsocketfactory.createSocket(workerIP,
			// workerTCPPort);registryIP

			/********** JobResult Information Transfer **********/	
			String filePath = "masterJobs/" + clientID;
			File file = new File(filePath);
			File fileList[] = file.listFiles();
			List<File> fileLists = Arrays.asList(fileList);
			String jobResultXML = null;
			try {
				JAXBContext context = JAXBContext.newInstance(JobResultObject.class);
				Marshaller m = context.createMarshaller();
			    m.setProperty(Marshaller.JAXB_FORMATTED_OUTPUT, true);

			    JobResultObject jobResult = new JobResultObject();
			    
			    jobResult.setFrom("Master");
			    jobResult.setType("SendJobResult");
			    jobResult.setTimestamp(getTime());
			    jobResult.setJob(new JobObject());
			    jobResult.setNumberOfOutputFiles(fileLists.size());
			    
			    if (!fileLists.isEmpty()) {
			    	Iterator<File> iter = fileLists.iterator();
			    	while (iter.hasNext()) {
						File file1 = iter.next();
						if (file1.isFile()) {
							FileObject outputFile = new FileObject();
							outputFile.setFileName(file1.getName());
							outputFile.setFileSize(file1.length());
							jobResult.addFile(outputFile);
						}
					}
			    }
			    
			    ByteArrayOutputStream baos = new ByteArrayOutputStream();
			    m.marshal(jobResult, baos);
			    jobResultXML = baos.toString();
			    
			} catch (JAXBException e1) {
				// TODO Auto-generated catch block
				e1.printStackTrace();
			}
			System.out.println("DEBUG - XML Result:");
			System.out.println(jobResultXML);
			
			dataOut.writeUTF(jobResultXML);
			System.out.println("SYSINFO - Send a Job Information.");
			
			/********** File Transfer **********/
			// NTD: file filter
			Iterator<File> iter = fileLists.iterator();
			while(iter.hasNext()){
				File resultFile = iter.next();
				if(resultFile.isFile()){
					System.out.println("DEBUG - File name:" +  resultFile.getName());

					// send the file content
					DataInputStream fileIn = new DataInputStream(new BufferedInputStream(new FileInputStream(resultFile)));
					int bufferSize = 8192;
					byte[] buf = new byte[bufferSize];
					int length = 0;
					while((length = fileIn.read(buf))!=-1) {
						dataOut.write(buf, 0, length);
					}
					dataOut.flush();
				}
				else{
					System.out.println("SYSINFO - "+file.getName()+" is a directory.");
				}
				System.out.println("DEBUG - Send a output file.");
			}
			System.out.println("SYSINFO - Send a job Result.");
		} catch (Exception e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		} 

	}
	
	public void createXML(Object o, String objectName, String objectClass, String from, String type) {
	}

    public boolean killJobInWorkerJobList(String clientID, int jobID) {
//        //find the job which need to be killed in jobPackageInfo and change its status into "COMPLETED"
//        for (int i = 0; i < jobPackageInfo.size(); i++) {
//            JobsPackage jp = jobPackageInfo.elements().nextElement();
//            if (jp.getClientID().equals(clientID)) {
//                for (int j = 0; j < jp.getJobs().size(); j++) {
//                    if (jp.getJobs().get(j).getJobID() == jobID) {
//                        jp.getJobs().get(j).setStatus(COMPLETED);
//                        System.out.println("DEBUG - Find the job needed to be kill in jobPackageInfo and change its status into COMPLETED");
//                    }
//                }
//            }
//        }
        
        for (int i = 0; i < workerList.size(); i++) {
            WorkerObject worker = workerList.get(i);
            for (int j = 0; j < worker.getJobsList().size(); j++) {
                JobObject job = worker.getJobFromJobsList(j);
                if (job.getClientID().equals(clientID) && job.getJobID() == jobID) {
                    System.out.println("DEBUG - Find the job needed to be kill");
                    //when j==0 means the job is running in worker side
                    if (j == 0) {
                        String workerIP = worker.getIp();
                        int workerPort = worker.getPort();
                        //send request to kill the job in worker
                        sendKillJobToWorker(workerIP, workerPort, jobID);
                        System.out.println("DEBUG - Worker("+i+") List Size: "+workerList.get(i).getJobsList().size());
                        workerList.get(i).removeJobsList(0);
                        System.out.println("DEBUG - Worker("+i+") List Size: "+workerList.get(i).getJobsList().size());
                        workerList.get(i).setStatus(AVAILABLE);
                        System.out.println("DEBUG - Worker("+i+") Status: "+workerList.get(i).getStatus());
                        System.out.println("DEBUG - Job is killed in worker side");
                    } //when j!=0 means the job is in worker job list
                    else {
                        workerList.get(i).removeJobsList(j);
                         System.out.println("DEBUG - Job is killed in worker job list");

                    }
                  //  return true;
                }
            }
        }
        //find the job which need to be killed in jobPackageInfo and change its status into "COMPLETED"
        ArrayList<JobObject> jobs = jobPackageInfo.get(clientID).getJobs();
        Iterator<JobObject> jobIter = jobs.iterator();
        while (jobIter.hasNext()) {
            JobObject job = jobIter.next();
            if (job.getJobID() == jobID) {
                //jobPackageInfo.get(clientID).getJobs().get(p).setStatus(COMPLETED);
            	job.setStatus(FAILED);
            	job.setJobEndTime(getTime());
            	if (jobPackageInfo.get(clientID).getNumberOfJobsCompleted() + jobPackageInfo.get(clientID).getNumberOfJobsFailed()
            			== jobPackageInfo.get(clientID).getNumberOfJobs()) {
            		jobPackageInfo.get(clientID).setEndTime(getTime());
            	}
                return true;
//                jobPackageInfo.get(clientID).getJobs().get(p).setStatus(FAIL);
//                jobPackageInfo.get(clientID).deleteNumberOfJobsCompleted();
            }
        }
     
        System.out.println("SYSINFO - Can not find the job in Master side");
        return false;
    }

//    public boolean killJobInWorkerSide(String clientID, int jobID) {
//        for (int i = 0; i < jobPackageInfo.size(); i++) {
//            JobsPackage jp = jobPackageInfo.elements().nextElement();
//            if (jp.getClientID().equals(clientID)) {
//                for (int j = 0; j < jp.getJobs().size(); j++) {
//                    if (jp.getJobs().get(j).getJobID() == jobID) {
//                        int workerID = jp.getJobs().get(j).getAllocateWorkerID();
//                        for(int p=0; p<workerList.size(); p++){
//                            if(workerList.get(p).getWorkerID()== workerID){
//                                sendKillJobToWorker(workerList.get(p).getIp(),workerList.get(p).getPort(),jobID);
//                                return true;
//                            }
//                        }
//                    }
//                }
//            }
//        }
//        return false;
//    }

    public void sendKillJobToWorker(String workerIP, int workerPort, int jobID){
        try {
            Socket workerSocket = new Socket(workerIP, workerPort);
            DataOutputStream dos = new DataOutputStream(workerSocket.getOutputStream());
            String killJob = createKillJobXML(jobID);

            dos.writeUTF(killJob);
           // dos.writeUTF("<killJob>"+"<jobID>"+jobID+"</jobID>"+"</killJob>");
        } catch (Exception ex) {
            ex.printStackTrace();
        } 
    }

    public String createKillJobXML(int aJobID) throws Exception {
		DocumentBuilderFactory dbfactory = DocumentBuilderFactory.newInstance();
		DocumentBuilder db = null;
		//jp.setClientID(ClientID);
		try {
			db = dbfactory.newDocumentBuilder();
		} catch (ParserConfigurationException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}

		Document outDoc = db.newDocument();
		outDoc.setXmlVersion("1.0");
		Element outRoot = outDoc.createElement("Request");
		outRoot.setAttribute("type", "KillJob");
		outRoot.setAttribute("from", "Master");
		outRoot.setAttribute("version", "1");
		
		Element jobIDElement = outDoc.createElement("jobID");
		jobIDElement.appendChild(outDoc.createTextNode(Integer.toString(aJobID)));
		outRoot.appendChild(jobIDElement);

	    outDoc.appendChild(outRoot);

	    String xmlStr = Util.docToString(outDoc);
	    System.out.println("DEBUG - XML Kill Job: "+xmlStr);

	    return xmlStr;
    }

}

class WorkerHandlingThread extends Thread {
	private WorkerObject worker;
	private MasterService masterService;

	public WorkerHandlingThread(WorkerObject worker, MasterService masterService) {
		this.worker = worker;
		this.masterService = masterService;
	}

	public void run() {
		System.out.println("SYSINFO - Worker Handling Service started.");
		System.out.println("DEBUG - ip = " + worker.getIp());
		System.out.println("DEBUG - port = " + worker.getPort());
		System.out.println("DEBUG - price = " + worker.getPrice());
		System.out.println("DEBUG - load = " + worker.getLoad());
		System.out.println("DEBUG - time = " + worker.getRegTime());
		masterService.workerQueueHandling(worker);
	}
}

class RescheduleWorkerJobs extends Thread {
	private ArrayList<JobObject> jobList;
	ArrayList<WorkerObject> workerList;
	private MasterService masterService;
	public RescheduleWorkerJobs(MasterService masterService, ArrayList<WorkerObject> workerList, ArrayList<JobObject> jobList) {
		this.masterService = masterService;
		this.jobList = jobList;
		this.workerList = workerList;
	}

	public void run() {
		rescheduleJobs();
	}

	private void rescheduleJobs() {
		Iterator<JobObject> iter = jobList.iterator();
		while (iter.hasNext()) {
			JobObject job = iter.next();
			//increment Priority
			job.setPriority(job.getPriority() + 1);
			//reassign Job
			int estimatedExecTime = job.getTime();
			int deadline = job.getExtraTime() + estimatedExecTime;
			JobsPackage jp = masterService.jobPackageInfo.get(job.getClientID());
			int numberOfJobsInPackage = jp.getNumberOfJobs();
			int budget = jp.getBudget();
			int budgetPerJob = budget/numberOfJobsInPackage;
			String type = jp.getType();
			int i=0;
			if (type.matches("TIME")) {
				Collections.sort(workerList, new SortWorkers("TIME", "ACS"));
				while (i < workerList.size()-1 && masterService.checkCost(workerList.get(i), job, budgetPerJob))
					i++;
			}
			else {
				Collections.sort(workerList, new SortWorkers("PRICE", "ACS"));
				while (i < workerList.size()-1 && masterService.checkTime(workerList.get(i), job, deadline))
					i++;
			}
			workerList.get(i).setJobsList(masterService.addJobToList(job, workerList.get(i).getJobsList(), jp.getDeadline()));
			workerList.get(i).setLoad(workerList.get(i).getLoad() + job.getTime());

		}

	}
}

class ReceiveAJobResultThread extends Thread {
	Socket workerSocket;
	private MasterService masterService;

	public ReceiveAJobResultThread(Socket workerSocket, MasterService masterService) {
		this.workerSocket = workerSocket;
		this.masterService = masterService;
	}

	public void run() {
		masterService.receiveAJobResult(workerSocket);
	}
}
/*class AssignEachJobThread extends Thread {
	Socket socket;
	JobObject job;
	DataInputStream dataIn;
	DataOutputStream dataOut;
	MasterService masterService;
	ArrayList<String> resultList = new ArrayList<String>();

	public AssignEachJobThread(Socket socket, JobObject job, ArrayList<String> resultList) {
		this.socket = socket;
		this.job = job;
		this.resultList = resultList;

		masterService = new MasterService();
		try {
			InputStream in = socket.getInputStream();
			dataIn = new DataInputStream(in);

			OutputStream out = socket.getOutputStream();
			dataOut = new DataOutputStream(out);
		} catch (IOException e) {System.out.println("IO:"+e.getMessage());
		}
	}

	public void run() {

		try {
			String jobRequest = masterService.createJobRequestXML(job);
			dataOut.writeUTF(jobRequest);
			System.out.println("SYSINFO - Send a job request.");

			String filePath = job.getExecuteFilePath()+"/"+job.getExecuteFileName()+".class";
            File file = new File(filePath);
            System.out.println("DEBUG - File length:" + (int) file.length());

            // send the file content
            DataInputStream fileIn = new DataInputStream(new BufferedInputStream(new FileInputStream(filePath)));
            int bufferSize = 8192;
            byte[] buf = new byte[bufferSize];

            int length = 0;
            while((length = fileIn.read(buf))!=-1) {
            	dataOut.write(buf, 0, length);
            }
            dataOut.flush();

		} catch (Exception e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		} finally {
			if(socket!=null) {
				try {
					socket.close();
					System.out.println("SYSINFO - Client to Worker TCP connection is close.");
				} catch(IOException e){close failed}
			}
		}

		// receive a job reply from Worker
		try {
			InputStream in = socket.getInputStream();
			DataInputStream dataIn = new DataInputStream(in);
			String jobReply = dataIn.readUTF();
			System.out.println("SYSINFO - Received a list reply.");
			System.out.println("SYSINFO - Reply XML - "+jobReply);

			// process the list reply and restore the available worker list
			Document inDoc = Util.stringToDoc(jobReply);
			Element inRoot = inDoc.getDocumentElement();

			if (inRoot.getAttribute("type").equals("JobReply")) {
				System.out.println("SYSINFO - The reply is JobReply.");
				NodeList nodeList = inRoot.getElementsByTagName("Result");
//				if(!nodeList.getLength()==0)()
				for (int i = 0; i < nodeList.getLength(); i++) {
					Node node = nodeList.item(i);
					String result = node.getNodeValue();

						System.out.println("SYSINFO - Worker("+i+") Result: " + result);
						resultList.add(result);
				}
			} else {
				System.out.println("SYSINFO - The reply is not JobReply. No action will be performed.");
			}
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		} catch (ParserConfigurationException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		} catch (SAXException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}
}
 */
